package com.jerry.frame.rpc.registry;

import com.jerry.frame.rpc.common.NamedThreadFactory;
import io.netty.util.internal.ConcurrentSet;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * Created by zjh on 2017/8/7.
 */

//注册管理服务，用来查询，增删，订阅服务注册，以及注册服务监听，相当于服务注册的管理器
public abstract class AbstractRegistryService implements RegistryService {
    //待注册服务的队列
    private final LinkedBlockingQueue<RegistryMeta> queue = new LinkedBlockingQueue<RegistryMeta>();
    //注册服务的状态跟踪
    //使用protected是为了让子类可以使用该属性，只要在同一个包下面
    protected final ConcurrentMap<RegistryMeta, RegistryState> registryMetaMap = new ConcurrentHashMap();
    //订阅服务集合与监听的映射
    private final ConcurrentMap<RegistryMeta.ServiceMeta, CopyOnWriteArrayList<NotifyListener>> listConcurrentMap = new ConcurrentHashMap<RegistryMeta.ServiceMeta, CopyOnWriteArrayList<NotifyListener>>();
    //订阅的服务集合
    protected final ConcurrentSet<RegistryMeta.ServiceMeta> subscribeSet = new ConcurrentSet();
    //已经注册的成功的服务及提供者信息
    private final ConcurrentMap<RegistryMeta.ServiceMeta, RegistryValue> registries =
            new ConcurrentHashMap();
    private final ExecutorService registerExecutor =
            Executors.newSingleThreadExecutor(new NamedThreadFactory("register.executor"));
    private final ExecutorService localRegisterWatchExecutor =
            Executors.newSingleThreadExecutor(new NamedThreadFactory("local.register.watch.executor"));
    //注册器关闭标识
    private final AtomicBoolean shutDown = new AtomicBoolean(false);

    private final ConcurrentMap<RegistryMeta.Address, CopyOnWriteArrayList<OfflineListener>> offlineListenerConcurrentMap = new ConcurrentHashMap<RegistryMeta.Address, CopyOnWriteArrayList<OfflineListener>>();

    public void register(RegistryMeta meta) {
        queue.add(meta);
    }

    public void unregister(RegistryMeta registryMeta) {
        //如果注册服务队列里不存在该服务的注册，则说明服务已经被注册
        if (!queue.remove(registryMeta)) {
            //清除已注册服务状态
            registryMetaMap.remove(registryMeta);
            doUnregister(registryMeta);
        }
    }


    protected abstract void doUnregister(RegistryMeta registryMeta);

    //订阅，并且绑定通知的监听
    public void subscribe(RegistryMeta.ServiceMeta serviceMeta, NotifyListener notifyListener) {
        CopyOnWriteArrayList<NotifyListener> listeners = listConcurrentMap.get(serviceMeta);
        if (listeners == null) {

            listeners = new CopyOnWriteArrayList<NotifyListener>();
            listConcurrentMap.put(serviceMeta, listeners);
        }
        listeners.add(notifyListener);
        doSubscribe(serviceMeta);
    }

    //通知服务提供者事件
    public void notify(RegistryMeta.ServiceMeta serviceMeta, long version, NotifyListener.NotifyEvent notifyEvent, RegistryMeta... arry) {
        if (arry == null || arry.length == 0) {
            return;
        }
        RegistryValue registryValue = registries.get(serviceMeta);
        if(registryValue==null){
            RegistryValue newRegistryValue = new RegistryValue();
            registryValue = registries.putIfAbsent(serviceMeta,newRegistryValue);
             if(registryValue==null){
                 registryValue = newRegistryValue;
             }
        }
        //以状态作为锁操作的结果标识，用来避免与加锁对象无关操作加入加锁操作中，导致锁操作时间太长。
        boolean needNotify =false;
        Lock writeLock = registryValue.reentrantReadWriteLock.writeLock();
        writeLock.lock();
        try{
            if(version > registryValue.version){
                if(notifyEvent.equals(NotifyListener.NotifyEvent.CHILD_ADDED)){
                    for(RegistryMeta registryMeta:arry) {
                        registryValue.registryMetas.add(registryMeta);
                    }
                }else if(notifyEvent.equals(NotifyListener.NotifyEvent.CHILD_REMOVED)){
                    for(RegistryMeta registryMeta:arry) {
                        registryValue.registryMetas.remove(registryMeta);
                    }

                }
                needNotify =true;

            }
        }finally {
            writeLock.unlock();
        }
        //根据锁定操作的结果标识在锁块外继续其他操作
        if(needNotify){
            CopyOnWriteArrayList<NotifyListener> copyOnWriteArrayList = listConcurrentMap.get(serviceMeta);
            for(RegistryMeta registryMeta :arry) {
                for (NotifyListener notifyListener : copyOnWriteArrayList) {
                    notifyListener.notify( registryMeta,notifyEvent);
                }
            }
        }
    }


    public abstract void doSubscribe(RegistryMeta.ServiceMeta serviceMeta);

    //地址链接中断离线
    public void offline(RegistryMeta.Address address) {
        //remove方法可以移除的同时，返回移除的value的引用
        CopyOnWriteArrayList<OfflineListener> listeners = offlineListenerConcurrentMap.remove(address);
        if (listeners != null) {
            for (OfflineListener l : listeners) {
                l.offline();
            }
        }
    }

    //添加地址链接离线监听
    public void offlineListing(RegistryMeta.Address address, OfflineListener offlineListener) {
//获取地址的离线监听列表
        CopyOnWriteArrayList<OfflineListener> offlineListeners = offlineListenerConcurrentMap.get(address);
        if (offlineListeners == null) {
            //如果为空则创建一个新的
            CopyOnWriteArrayList newOfflineListeners = new CopyOnWriteArrayList<OfflineListener>();
            //将新的加入的map
            //putIfAbsent方法，如果当前key存在不为空value，则不插入新的，而返回旧的value，如果为空则插入新的，并且返回新的引用，
            //并发情况下，保证map中取出的value是最新的值。
            offlineListeners = offlineListenerConcurrentMap.putIfAbsent(address, newOfflineListeners);
            //如果监听列表已经不为空，则直接向当前不为空列表加入监听，跳过下面
            if (offlineListeners == null) {
                //如果监听列表为空，则向新的列表加入监听
                offlineListeners = newOfflineListeners;
            }
        }
        offlineListeners.addIfAbsent(offlineListener);

    }

    public Collection<RegistryMeta> lookup(RegistryMeta.ServiceMeta serviceMeta) {
        RegistryValue value = registries.get(serviceMeta);
        if (value == null) {
            return new ArrayList<RegistryMeta>();
        }
        //获取对象的读锁
        final Lock lock = value.reentrantReadWriteLock.readLock();
        lock.lock();
        try {
            return new ArrayList(value.registryMetas);
        } finally {
            lock.unlock();
        }
    }

    public boolean isShutdown() {
        return shutDown.get();
    }

    //停止注册服务中心
    public void shutdownGracefully() {
        if (!shutDown.getAndSet(true)) {
            //停止注册
            registerExecutor.shutdown();
            //停止监听服务
            localRegisterWatchExecutor.shutdown();
            try {
                destroy();
            } catch (Exception ignored) {

            }
        }
    }

    public abstract void destroy();

    protected enum RegistryState {
        PREPARE,
        DONE
    }


    public AbstractRegistryService() {
//轮询队列，将队列中需要注册的服务进行注册
        registerExecutor.execute(new Runnable() {
            public void run() {
                while (!shutDown.get()) {
                    RegistryMeta registryMeta = null;
                    try {
                        registryMeta = queue.take();
                        //先置为准备状态，避免在调用该服务时，过滤未注册完成的服务
                        registryMetaMap.put(registryMeta, RegistryState.PREPARE);
                        doRegisterService(registryMeta);
                    } catch (Exception e) {
                        //处理过程中失败，则将注册的服务重新放回队列中
                        if (registryMeta != null) {
                            queue.add(registryMeta);
                        }
                    }
                }
            }
        });
        //同时对注册的服务的节点每三秒进行监控
        localRegisterWatchExecutor.execute(new Runnable() {
            public void run() {
                while (!shutDown.get()) {

                    try {
                        Thread.sleep(3000L);
                        doCheckRegisterNodeStatus();
                    } catch (Exception e) {


                    }

                }
            }
        });
    }

    public abstract void doRegisterService(final RegistryMeta registryMeta)throws Exception;

    public abstract void doCheckRegisterNodeStatus();

    //封装服务对应提供者的信息
    public static class RegistryValue {
        private long version;
        private final Set<RegistryMeta> registryMetas = new HashSet<RegistryMeta>();
        private final ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();


    }

}
